package com.sunl.mall.ali.mq.consumer;

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.batch.BatchMessageListener;
import com.aliyun.openservices.ons.api.bean.BatchConsumerBean;
import com.aliyun.openservices.ons.api.bean.Subscription;
import com.sunl.mall.ali.mq.api.consumer.RocketMqBatchMessageListener;
import com.sunl.mall.ali.mq.model.RocketMqMessageBody;
import com.sunl.mall.ali.mq.model.RocketMqMessageContext;
import com.sunl.mall.ali.mq.topic.SubscribeTopic;
import com.sunl.mall.ali.mq.util.MqContextUtil;
import com.sunl.mall.ali.mq.util.MqUtil;
import com.sunl.mall.common.api.topic.MqTopic;
import com.sunl.mall.common.api.util.ExceptionUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class RocketMqBatchConsumerBean extends BatchConsumerBean implements SubscriptionRelationChecker, SubscriptionLogger {
    private final Logger log = LoggerFactory.getLogger(RocketMqBatchConsumerBean.class);
    private String consumerGroupId;

    @Override
    public void start() {
        this.consumerGroupId = MqUtil.getConsumerId(this.getProperties());
        log.info("########## Start to create batchConsumer. consumerGroupId=[{}]", this.consumerGroupId);
        this.check(this.consumerGroupId);
        super.start();
    }

    @Override
    public void subscribe(final String topic, final String tags, final BatchMessageListener listener) {
        log.info("########## Start to subscribe topic. consumerGroupId=[{}] topic=[{}] tags=[{}]", this.consumerGroupId, topic, tags);
        super.subscribe(topic, tags, listener);
    }

    /**
     * 设置订阅关系
     * 1.将自定义 listener 转成 sdk中的 {@link MessageListener}
     * 2.调用父类中设置订阅关系逻辑
     *
     * @param subscriptionTable
     * @param <T>
     */
    public <T> void setSubscriptionRelation(final List<RocketMqBatchMessageListener<T>> subscriptions) {
        final Map<Subscription, BatchMessageListener> table = new HashMap<>(subscriptions.size());
        for (RocketMqBatchMessageListener<T> subscription : subscriptions) {
            MqTopic rockMqTopic = subscription.subscribe();
            if (rockMqTopic == null) {
                throw new IllegalArgumentException("MqConsumer[" + subscription.getClass() + "] subscribe topic must not null, and must implement method[subscribe]");
            }
            // 将自定义listener转成sdk中的listener
            BatchMessageListener exist = table.putIfAbsent(new SubscribeTopic(rockMqTopic), ((originMsgs, originContext) -> {
                final List<RocketMqMessageBody<T>> records = new ArrayList<>();
                final List<RocketMqMessageContext> contexts = new ArrayList<>();
                try {
                    // 循环处理消息体
                    for (int i = originMsgs.size() - 1; i >= 0; i--) {
                        Message originMsg = originMsgs.get(i);
                        RocketMqMessageContext context = MqContextUtil.getMessageContext(originMsg);
                        RocketMqMessageBody<T> record = subscription.deserialize(originMsg.getBody(), context);
                        this.logDeserializeResult(context, consumerGroupId, record);
                        records.add(record);
                        contexts.add(context);
                    }
                    // 业务批量处理消息
                    return subscription.call(records, contexts);
                } catch (Throwable throwable) {
                    Object[] params = {records.size(), this.consumerGroupId, records, contexts, ExceptionUtil.getAsString(throwable)};
                    log.error("Mq batchConsumer message Exception: message size:[{}], consumerGroupId:[{}], records:[{}], contexts[{}], Exception:[{}]", params);
                    return Action.ReconsumeLater;
                }
            }));
            if (exist != null) {
                throw new IllegalArgumentException("MqConsumer[" + subscription.getClass() + "] subscribe topic[" + rockMqTopic.getTopic() + "] tag[" + rockMqTopic.getExpression() + "] is exist!");
            }
        }
        // 设置订阅关系
        super.setSubscriptionTable(table);
    }
}
